//! # 连接池管理实现
//! 
//! 提供通用连接池的具体实现，支持连接验证、自动恢复和统计监控。
//! 采用异步设计，适用于高并发场景。

use super::traits::{ConnectionPool, ConnectionFactory, PooledConnection, PoolStats, ConnectionState};
use rustcloud_core::{ServiceResult, ServiceError};
use std::sync::Arc;
use std::collections::HashMap;
use std::time::{Duration, Instant};
use tokio::sync::{RwLock, Semaphore, Notify};
use async_trait::async_trait;

/// 通用连接池实现
pub struct GenericConnectionPool<T> {
    /// 连接池配置
    config: ConnectionPoolConfig,
    /// 连接工厂
    factory: Arc<dyn ConnectionFactory<T>>,
    /// 可用连接
    available: Arc<RwLock<Vec<PooledConnection<T>>>>,
    /// 使用中的连接
    in_use: Arc<RwLock<HashMap<String, PooledConnection<T>>>>,
    /// 信号量（控制最大连接数）
    semaphore: Arc<Semaphore>,
    /// 统计信息
    stats: Arc<RwLock<PoolStats>>,
    /// 通知器（用于连接可用通知）
    notify: Arc<Notify>,
}

/// 连接池配置
#[derive(Debug, Clone)]
pub struct ConnectionPoolConfig {
    /// 最大连接数
    pub max_connections: u32,
    /// 最小连接数
    pub min_connections: u32,
    /// 连接超时（秒）
    pub connection_timeout_seconds: u64,
    /// 空闲超时（秒）
    pub idle_timeout_seconds: u64,
    /// 验证间隔（秒）
    pub validation_interval_seconds: u64,
    /// 是否启用连接验证
    pub validation_enabled: bool,
}

impl Default for ConnectionPoolConfig {
    fn default() -> Self {
        Self {
            max_connections: 100,
            min_connections: 10,
            connection_timeout_seconds: 30,
            idle_timeout_seconds: 300,
            validation_interval_seconds: 60,
            validation_enabled: true,
        }
    }
}

impl<T: Send + 'static> GenericConnectionPool<T> {
    /// 创建新的连接池
    pub fn new(
        config: ConnectionPoolConfig,
        factory: Arc<dyn ConnectionFactory<T>>,
    ) -> Self {
        let semaphore = Arc::new(Semaphore::new(config.max_connections as usize));
        let notify = Arc::new(Notify::new());
        
        Self {
            config,
            factory,
            available: Arc::new(RwLock::new(Vec::new())),
            in_use: Arc::new(RwLock::new(HashMap::new())),
            semaphore,
            stats: Arc::new(RwLock::new(PoolStats::default())),
            notify,
        }
    }

    /// 创建新连接
    async fn create_new_connection(&self) -> ServiceResult<PooledConnection<T>> {
        let connection = self.factory.create_connection().await?;
        let mut pooled = PooledConnection::new(connection);
        pooled.mark_used();
        
        // 更新统计
        {
            let mut stats = self.stats.write().await;
            stats.create_count += 1;
            stats.total_connections += 1;
            stats.active_connections += 1;
        }
        
        Ok(pooled)
    }

    /// 验证连接有效性
    async fn validate_connection_internal(&self, connection: &mut PooledConnection<T>) -> bool {
        if !self.config.validation_enabled {
            return true;
        }
        
        let is_valid = self.factory.validate_connection(&connection.connection).await;
        if !is_valid {
            let mut stats = self.stats.write().await;
            stats.validation_failures += 1;
        }
        is_valid
    }

    /// 销毁连接
    async fn destroy_connection_internal(&self, pooled_connection: PooledConnection<T>) -> ServiceResult<()> {
        self.factory.destroy_connection(pooled_connection.connection).await?;
        
        // 更新统计
        {
            let mut stats = self.stats.write().await;
            stats.destroy_count += 1;
            stats.total_connections = stats.total_connections.saturating_sub(1);
            stats.active_connections = stats.active_connections.saturating_sub(1);
        }
        
        Ok(())
    }
}

#[async_trait]
impl<T: Send + Sync + 'static> ConnectionPool<T> for GenericConnectionPool<T> {
    async fn acquire(&self) -> ServiceResult<PooledConnection<T>> {
        // 更新统计
        {
            let mut stats = self.stats.write().await;
            stats.acquire_count += 1;
        }

        // 获取信号量许可（控制最大连接数）
        let _permit = match tokio::time::timeout(
            Duration::from_secs(self.config.connection_timeout_seconds),
            self.semaphore.acquire()
        ).await {
            Ok(permit) => permit,
            Err(_) => {
                return Err(ServiceError::Timeout(
                    format!("获取连接超时，超过 {} 秒", self.config.connection_timeout_seconds)
                ));
            }
        };

        // 首先尝试从可用连接中获取
        let connection = {
            let mut available = self.available.write().await;
            if let Some(mut pooled) = available.pop() {
                // 验证连接是否仍然有效
                if self.validate_connection_internal(&mut pooled).await {
                    pooled.mark_used();
                    
                    // 更新统计
                    {
                        let mut stats = self.stats.write().await;
                        stats.idle_connections = stats.idle_connections.saturating_sub(1);
                        stats.active_connections += 1;
                    }
                    
                    Some(pooled)
                } else {
                    // 连接无效，销毁它
                    if let Err(e) = self.destroy_connection_internal(pooled).await {
                        tracing::warn!("销毁无效连接时出错: {}", e);
                    }
                    None
                }
            } else {
                None
            }
        };

        // 如果没有可用连接，创建新连接
        let connection = if let Some(conn) = connection {
            conn
        } else {
            self.create_new_connection().await?
        };

        // 将连接添加到使用中列表
        {
            let mut in_use = self.in_use.write().await;
            in_use.insert(connection.id.clone(), connection.clone());
        }

        Ok(connection)
    }

    async fn release(&self, connection_id: &str) -> ServiceResult<()> {
        // 更新统计
        {
            let mut stats = self.stats.write().await;
            stats.release_count += 1;
        }

        // 从使用中列表移除
        let pooled_connection = {
            let mut in_use = self.in_use.write().await;
            in_use.remove(connection_id)
        };

        if let Some(mut pooled) = pooled_connection {
            // 标记为空闲
            pooled.mark_idle();
            
            // 验证连接是否仍然有效
            let is_valid = self.validate_connection_internal(&mut pooled).await;
            
            if is_valid {
                // 将连接放回可用列表
                {
                    let mut available = self.available.write().await;
                    available.push(pooled);
                }
                
                // 更新统计
                {
                    let mut stats = self.stats.write().await;
                    stats.active_connections = stats.active_connections.saturating_sub(1);
                    stats.idle_connections += 1;
                }
                
                // 通知等待的获取者
                self.notify.notify_one();
            } else {
                // 连接无效，销毁它
                if let Err(e) = self.destroy_connection_internal(pooled).await {
                    tracing::warn!("销毁无效连接时出错: {}", e);
                }
                
                // 释放信号量许可
                drop(_permit);
            }
            
            Ok(())
        } else {
            Err(ServiceError::InternalError(
                format!("连接 {} 不存在于使用中列表中", connection_id)
            ))
        }
    }

    async fn validate_connection(&self, connection_id: &str) -> ServiceResult<bool> {
        let in_use = self.in_use.read().await;
        if let Some(pooled) = in_use.get(connection_id) {
            let is_valid = self.factory.validate_connection(&pooled.connection).await;
            Ok(is_valid)
        } else {
            // 如果不在使用中列表，检查是否在可用列表中
            let available = self.available.read().await;
            for pooled in available.iter() {
                if pooled.id == connection_id {
                    let is_valid = self.factory.validate_connection(&pooled.connection).await;
                    return Ok(is_valid);
                }
            }
            
            Err(ServiceError::InternalError(
                format!("连接 {} 不存在", connection_id)
            ))
        }
    }

    async fn get_stats(&self) -> PoolStats {
        self.stats.read().await.clone()
    }

    async fn initialize(&self) -> ServiceResult<()> {
        let mut available = self.available.write().await;
        
        // 预创建最小连接数
        for i in 0..self.config.min_connections {
            match self.factory.create_connection().await {
                Ok(connection) => {
                    let pooled = PooledConnection::new(connection);
                    available.push(pooled);
                    
                    // 更新统计
                    {
                        let mut stats = self.stats.write().await;
                        stats.total_connections += 1;
                        stats.idle_connections += 1;
                        stats.create_count += 1;
                    }
                }
                Err(e) => {
                    tracing::warn!("初始化连接池时创建连接 {} 失败: {}", i, e);
                    return Err(e);
                }
            }
        }
        
        tracing::info!(
            "连接池初始化完成，预创建了 {} 个连接", 
            self.config.min_connections
        );
        
        Ok(())
    }

    async fn shutdown(&self) -> ServiceResult<()> {
        // 关闭所有可用连接
        let mut available = self.available.write().await;
        for pooled in available.drain(..) {
            if let Err(e) = self.factory.destroy_connection(pooled.connection).await {
                tracing::warn!("关闭连接池时销毁连接失败: {}", e);
            }
        }

        // 等待所有使用中的连接释放
        let in_use_count = self.in_use.read().await.len();
        if in_use_count > 0 {
            tracing::warn!("连接池关闭时仍有 {} 个连接在使用中", in_use_count);
        }

        // 更新统计
        {
            let mut stats = self.stats.write().await;
            stats.total_connections = 0;
            stats.idle_connections = 0;
            stats.active_connections = 0;
        }

        tracing::info!("连接池已关闭");
        Ok(())
    }

    async fn cleanup_expired(&self) -> ServiceResult<usize> {
        let idle_timeout = Duration::from_secs(self.config.idle_timeout_seconds);
        let mut expired_count = 0;

        {
            let mut available = self.available.write().await;
            
            // 保留未过期的连接，销毁过期的连接
            let expired_connections: Vec<PooledConnection<T>> = available
                .drain(..)
                .filter(|conn| conn.is_expired(idle_timeout))
                .collect();
            
            expired_count = expired_connections.len();
            
            // 销毁过期连接
            for pooled in expired_connections {
                if let Err(e) = self.factory.destroy_connection(pooled.connection).await {
                    tracing::warn!("清理过期连接时销毁连接失败: {}", e);
                }
            }
        }

        // 更新统计
        if expired_count > 0 {
            let mut stats = self.stats.write().await;
            stats.destroy_count += expired_count as u64;
            stats.total_connections = stats.total_connections.saturating_sub(expired_count);
            stats.idle_connections = stats.idle_connections.saturating_sub(expired_count);
            
            tracing::info!("清理了 {} 个过期连接", expired_count);
        }

        Ok(expired_count)
    }
}

/// 连接池构建器
pub struct ConnectionPoolBuilder<T> {
    config: ConnectionPoolConfig,
    factory: Option<Arc<dyn ConnectionFactory<T>>>,
}

impl<T: Send + Sync + 'static> ConnectionPoolBuilder<T> {
    /// 创建新的构建器
    pub fn new() -> Self {
        Self {
            config: ConnectionPoolConfig::default(),
            factory: None,
        }
    }

    /// 设置最大连接数
    pub fn max_connections(mut self, max_connections: u32) -> Self {
        self.config.max_connections = max_connections;
        self
    }

    /// 设置最小连接数
    pub fn min_connections(mut self, min_connections: u32) -> Self {
        self.config.min_connections = min_connections;
        self
    }

    /// 设置连接超时时间（秒）
    pub fn connection_timeout(mut self, seconds: u64) -> Self {
        self.config.connection_timeout_seconds = seconds;
        self
    }

    /// 设置空闲超时时间（秒）
    pub fn idle_timeout(mut self, seconds: u64) -> Self {
        self.config.idle_timeout_seconds = seconds;
        self
    }

    /// 设置验证间隔（秒）
    pub fn validation_interval(mut self, seconds: u64) -> Self {
        self.config.validation_interval_seconds = seconds;
        self
    }

    /// 启用或禁用连接验证
    pub fn validation_enabled(mut self, enabled: bool) -> Self {
        self.config.validation_enabled = enabled;
        self
    }

    /// 设置连接工厂
    pub fn with_factory<F: ConnectionFactory<T> + 'static>(mut self, factory: F) -> Self {
        self.factory = Some(Arc::new(factory));
        self
    }

    /// 构建连接池
    pub fn build(self) -> ServiceResult<GenericConnectionPool<T>> {
        let factory = self.factory.ok_or_else(|| {
            ServiceError::InternalError("必须提供连接工厂".to_string())
        })?;
        
        Ok(GenericConnectionPool::new(self.config, factory))
    }
}

impl<T: Send + Sync + 'static> Default for ConnectionPoolBuilder<T> {
    fn default() -> Self {
        Self::new()
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::sync::atomic::{AtomicBool, AtomicU32, Ordering};

    /// 模拟连接
    #[derive(Debug, Clone)]
    pub struct MockConnection {
        pub id: String,
        is_valid: Arc<AtomicBool>,
    }

    impl MockConnection {
        pub fn new(id: String) -> Self {
            Self {
                id,
                is_valid: Arc::new(AtomicBool::new(true)),
            }
        }

        pub fn set_invalid(&self) {
            self.is_valid.store(false, Ordering::Relaxed);
        }

        pub fn is_valid(&self) -> bool {
            self.is_valid.load(Ordering::Relaxed)
        }
    }

    /// 模拟连接工厂
    pub struct MockConnectionFactory {
        should_fail: Arc<AtomicBool>,
        connection_counter: Arc<AtomicU32>,
    }

    impl MockConnectionFactory {
        pub fn new() -> Self {
            Self {
                should_fail: Arc::new(AtomicBool::new(false)),
                connection_counter: Arc::new(AtomicU32::new(0)),
            }
        }

        pub fn set_should_fail(&self, fail: bool) {
            self.should_fail.store(fail, Ordering::Relaxed);
        }
    }

    #[async_trait]
    impl ConnectionFactory<MockConnection> for MockConnectionFactory {
        async fn create_connection(&self) -> ServiceResult<MockConnection> {
            if self.should_fail.load(Ordering::Relaxed) {
                return Err(ServiceError::InternalError("模拟连接创建失败".to_string()));
            }

            let id = self.connection_counter.fetch_add(1, Ordering::Relaxed);
            Ok(MockConnection::new(format!("mock_conn_{}", id)))
        }

        async fn validate_connection(&self, connection: &MockConnection) -> bool {
            connection.is_valid()
        }

        async fn destroy_connection(&self, _connection: MockConnection) -> ServiceResult<()> {
            // 模拟连接销毁
            tracing::debug!("销毁模拟连接");
            Ok(())
        }

        fn connection_type(&self) -> &str {
            "MockConnection"
        }
    }

    #[tokio::test]
    async fn test_connection_pool_basic_operations() {
        let config = ConnectionPoolConfig::default();
        let factory = Arc::new(MockConnectionFactory::new());
        let pool = GenericConnectionPool::new(config, factory.clone());

        // 初始化连接池
        pool.initialize().await.unwrap();

        // 获取连接
        let conn = pool.acquire().await.unwrap();
        assert_eq!(conn.connection.id, "mock_conn_0");
        assert_eq!(conn.state, ConnectionState::Active);

        // 验证统计信息
        let stats = pool.get_stats().await;
        assert_eq!(stats.active_connections, 1);
        assert_eq!(stats.acquire_count, 1);

        // 释放连接
        pool.release(&conn.id).await.unwrap();

        let stats = pool.get_stats().await;
        assert_eq!(stats.active_connections, 0);
        assert_eq!(stats.release_count, 1);
        assert_eq!(stats.idle_connections, 1);

        // 关闭连接池
        pool.shutdown().await.unwrap();
    }

    #[tokio::test]
    async fn test_connection_validation() {
        let config = ConnectionPoolConfig::default();
        let factory = Arc::new(MockConnectionFactory::new());
        let pool = GenericConnectionPool::new(config, factory.clone());

        pool.initialize().await.unwrap();

        // 获取连接
        let conn = pool.acquire().await.unwrap();
        
        // 验证有效连接
        let is_valid = pool.validate_connection(&conn.id).await.unwrap();
        assert!(is_valid);

        // 设置连接为无效
        conn.connection.set_invalid();
        
        // 验证无效连接
        let is_valid = pool.validate_connection(&conn.id).await.unwrap();
        assert!(!is_valid);

        pool.release(&conn.id).await.unwrap();
        pool.shutdown().await.unwrap();
    }

    #[tokio::test]
    async fn test_connection_factory_failure() {
        let config = ConnectionPoolConfig::default();
        let factory = Arc::new(MockConnectionFactory::new());
        let pool = GenericConnectionPool::new(config, factory.clone());

        pool.initialize().await.unwrap();

        // 设置工厂失败
        factory.set_should_fail(true);

        // 尝试获取连接应该失败
        let result = pool.acquire().await;
        assert!(result.is_err());

        pool.shutdown().await.unwrap();
    }

    #[tokio::test]
    async fn test_expired_connection_cleanup() {
        let mut config = ConnectionPoolConfig::default();
        config.idle_timeout_seconds = 0; // 立即过期
        let factory = Arc::new(MockConnectionFactory::new());
        let pool = GenericConnectionPool::new(config, factory.clone());

        pool.initialize().await.unwrap();

        // 获取并释放连接
        let conn = pool.acquire().await.unwrap();
        pool.release(&conn.id).await.unwrap();

        // 清理过期连接
        let cleaned_count = pool.cleanup_expired().await.unwrap();
        assert_eq!(cleaned_count, 1);

        pool.shutdown().await.unwrap();
    }
}